package com.dukk.espark.apps;

import com.alibaba.fastjson.JSONObject;
import com.dukk.espark.core.Engine;
import com.dukk.espark.core.SparkBase;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Iterator;

/**
 * javaRdd操作示例类
 * @version 2020-11-09
 * @author shengshi_feiyang@yeah.net
 */
@Engine(name = "JavaRDDApp")
public class JavaRDDApp extends SparkBase {

    private static final Logger logger = LoggerFactory.getLogger(JavaRDDApp.class);

    @Override
    public void doAi(JSONObject params) {

        //示例一: list转javaRDD
        String[] strDate = {"spark","java","es","python","scala","javaScripts","spark","java","es","python","scala","javaScripts"};
        JavaRDD<String> listJavaRDD = javaSparkContext.parallelize(Arrays.asList(strDate));
        listJavaRDD.foreachPartition((VoidFunction<Iterator<String>>)i->{ //foreachPartition遍历数据
            while (i.hasNext()){
                logger.info("list转Rdd遍历输出:{}", i.next());
            }
        });

        //示例二: map操作, 计算每个字符串的长度
        JavaRDD<Integer> mapJavaRdd = listJavaRDD.map((Function<String, Integer>)f->{
            return f.length();
        });

        mapJavaRdd.foreach((VoidFunction<Integer>)i->{ //foreach遍历数据
            logger.info("mapJavaRdd遍历输出:{}", i);
        });

        //示例三: filter过滤数据
        JavaRDD<String> filterJavaRdd = listJavaRDD.filter((Function<String, Boolean>)f->{
           if(f.equals("es")){
               return false;
           }
           return true;
        });

        filterJavaRdd.foreach((VoidFunction<String>)i->{ //foreach遍历数据
            logger.info("filterJavaRdd遍历输出:{}", i);
        });



    }
}
